/*  Pierwsze zadanie z zaliczeniowe z przedmiotu
        Kolumnowe i rozproszone bazy danych
    autor: Lukasz Marecik
           lm277561
*/

#include <cstdio>
//#include <iostream>
#include <cstdio>
#include <fcntl.h>
#include <vector>
#include <cmath>
#include <unordered_set>
#include <google/protobuf/text_format.h>
#include <google/protobuf/io/zero_copy_stream_impl.h>
#include "operations.pb.h"
#include "server.h"

using namespace std ;

#define TABLESIZE 2048

enum Type {
	INT = 1, DOUBLE = 2, BOOL = 3
} ;

struct Output_elem {
	void* column ;	// int* | double* | bool*
	Type type ;
	Output_elem(void* c = 0, Type t = INT) : column(c), type(t) {}
} ;

typedef vector<Output_elem> Output ;

class Op {
  protected:
  	Output output ;
  public:
	Output getOutput() {
		return output ;
	}
	// zwraca liczbe wierszy lub -1 jezeli juz koniec
	virtual int doit() = 0 ;
} ;

Server *server ;
Op* createOp(const Operation &operation) ;

class OpScan: public Op {
  private:
  	vector<int> column_number ;
  public:
  	OpScan(const ScanOperation &operation) {
//  		printf("tworze OpScan\n") ;
		int n = operation.column().size() ;
		for(int i=0 ; i<n ; i++) {
			switch(operation.type(i)) {
				case ScanOperation_Type_INT:
					output.push_back(Output_elem( malloc(sizeof(int32)*TABLESIZE), INT)) ;
					break ;
				case ScanOperation_Type_DOUBLE:
					output.push_back(Output_elem( malloc(sizeof(double)*TABLESIZE), DOUBLE)) ;
					break ;
				case ScanOperation_Type_BOOL:
					output.push_back(Output_elem( malloc(sizeof(bool)*TABLESIZE), BOOL)) ;
					break ;
				default:
					assert(false) ;
			}
			column_number.push_back(operation.column(i)) ;
		}
	}
	int doit() {
		int ret = 0 ;
		for(size_t i=0 ; i<output.size() ; i++) {
			switch(output[i].type) {
				case INT:
					ret = server->GetInts(column_number[i], TABLESIZE, (int32*) output[i].column) ;
					break ;
				case DOUBLE:
					ret = server->GetDoubles(column_number[i], TABLESIZE, (double*) output[i].column) ;
					break ;
				case BOOL:
					ret = server->GetByteBools(column_number[i], TABLESIZE, (bool*) output[i].column) ;
					break ;
			}
		}
		return (ret==0) ? -1 : ret ;
	}
} ;

class Exp {
  protected:
  	Output_elem output ;
  public:
  	Output_elem getOutputElem() {
  		return output ;
  	}
  	virtual void doit(int size) = 0 ;
} ;

class ColumnExp: public Exp {
  public:
  	ColumnExp(const Output &in, int nr) {
  		output = in[nr] ;
  	}
  	void doit(int) {}
} ;

/******* typeTraitsy rzutujace class na Type  **********/

template<class T>
struct toType {
	static Type X ;
} ;

template<>
struct toType<int32> {
	static Type X ;
} ;
Type toType<int32>::X = INT ;

template<>
struct toType<double> {
	static Type X ;
} ;
Type toType<double>::X = DOUBLE ;

template<>
struct toType<bool> {
	static Type X ;
} ;
Type toType<bool>::X = BOOL ;

/******************************************************/

template<class T>
class ConstantExp: public Exp {
  public:
  	ConstantExp(T x) {
//  		cout << "Tworze kolumne " << x << endl ;
  		T* data = new T[TABLESIZE] ;
  		for(int i=0 ; i<TABLESIZE ; i++) data[i] = x ;
  		output = Output_elem((void *) data, toType<T>::X) ;
  	}
  	void doit(int) {}
} ;

// result_type to typ wynikowy dodawania typu A i typu B,
template<class A, class B>
struct traits {
	typedef double result_type ;
} ;

template<>
struct traits<int32, int32> {
	typedef int32 result_type ;
} ;

// MAKRO tworzace szablon klasy wyrazen dwuargumentowych
#define TWO_ARGUMENTS_EXPRESSION(NAME, ACTION, RESULT, SPECIAL)    \
template<class A, class B>                                         \
class NAME: public Exp {                                           \
  private:                                                         \
  	RESULT *data ;                                                 \
  	A* sourceA ;                                                   \
  	B* sourceB ;                                                   \
  	Exp* a ;                                                       \
  	Exp* b ;                                                       \
  public:                                                          \
  	NAME(Exp* a_, Exp* b_) : a(a_), b(b_) {                        \
  		data = new RESULT[TABLESIZE] ;                             \
  		output = Output_elem((void *) data, toType< RESULT >::X) ; \
  		sourceA = (A*) a->getOutputElem().column ;                 \
  		sourceB = (B*) b->getOutputElem().column ;                 \
  	}                                                              \
  	void doit(int size) {                                          \
  		a->doit(size) ;                                            \
  		b->doit(size) ;                                            \
  		for(int i=0 ; i<size ; i++)                                \
  			data[i] = (SPECIAL sourceA[i] ACTION sourceB[i]) ;     \
  	}                                                              \
} ;


#define COMMA ,

// tworzenie szablonow
TWO_ARGUMENTS_EXPRESSION(AddExp,      +, typename traits<A COMMA B>::result_type, )
TWO_ARGUMENTS_EXPRESSION(SubtractExp, -, typename traits<A COMMA B>::result_type, )
TWO_ARGUMENTS_EXPRESSION(MultiplyExp, *, typename traits<A COMMA B>::result_type, )
TWO_ARGUMENTS_EXPRESSION(LowerExp, <, bool, )
TWO_ARGUMENTS_EXPRESSION(GreaterExp, >, bool, )
TWO_ARGUMENTS_EXPRESSION(EqualExp, ==, bool, )
TWO_ARGUMENTS_EXPRESSION(NotEqualExp, !=, bool, )
TWO_ARGUMENTS_EXPRESSION(OrExp, ||, bool, )
TWO_ARGUMENTS_EXPRESSION(AndExp, &&, bool, )
TWO_ARGUMENTS_EXPRESSION(DivideExp, /, double, (double) )

// makra instancjonujace szablony
#define TYPE_CHOISE_1(NAME, ACTION)                            \
if(t1 == INT    && t2 == INT)    {                             \
	/*printf("INT %s INT\n", #ACTION) ;*/                      \
	return new NAME<int32,int32> (a,b) ;                       \
}                                                              \
if(t1 == INT    && t2 == DOUBLE) {                             \
	/*printf("INT %s DOUBLE\n", #ACTION) ; */                  \
	return new NAME<int32,double>(a,b) ;                       \
}                                                              \
if(t1 == DOUBLE && t2 == INT)   {                              \
	/*printf("DOUBLE %s INT\n", #ACTION) ;*/                   \
	return new NAME<double,int32> (a,b) ;                      \
}                                                              \
if(t1 == DOUBLE && t2 == DOUBLE) {                             \
	/*printf("DOUBLE %s DOUBLE\n", #ACTION) ;*/                \
	return new NAME<double,double>(a,b) ;                      \
}

#define TYPE_CHOISE_2(NAME, ACTION)                            \
if(t1 == INT    && t2 == INT)    {                             \
	/*printf("INT %s INT\n", #ACTION) ; */                     \
	return new NAME<int32,int32> (a,b) ;                       \
}                                                              \
if(t1 == DOUBLE    && t2 == DOUBLE) {                          \
	/*printf("DOUBLE %s DOUBLE\n", #ACTION) ; */               \
	return new NAME<double,double>(a,b) ;                      \
}                                                              \
if(t1 == BOOL && t2 == BOOL)   {                               \
	/*printf("BOOL %s BOOL\n", #ACTION) ; */                   \
	return new NAME<bool,bool> (a,b) ;                         \
}

#define TYPE_CHOISE_3(NAME, ACTION)                            \
if(t1 == BOOL && t2 == BOOL)   {                               \
	/*printf("BOOL %s BOOL\n", #ACTION) ; */                   \
	return new NAME<bool,bool> (a,b) ;                         \
}

// MAKRO tworzace szablon klasy wyrazen jednoargumentowych
#define ONE_ARGUMENT_EXPRESSION(NAME, ACTION, RESULT)              \
template<class A>                                                  \
class NAME: public Exp {                                           \
  private:                                                         \
  	RESULT *data ;                                                 \
  	A* sourceA ;                                                   \
  	Exp* a ;                                                       \
  public:                                                          \
  	NAME(Exp* a_) : a(a_) {                                        \
  		data = new RESULT[TABLESIZE] ;                             \
  		output = Output_elem((void *) data, toType< RESULT >::X) ; \
  		sourceA = (A*) a->getOutputElem().column ;                 \
  	}                                                              \
  	void doit(int size) {                                          \
  		a->doit(size) ;                                            \
  		for(int i=0 ; i<size ; i++)                                \
  			data[i] = ACTION(sourceA[i]) ;                         \
  	}                                                              \
} ;

ONE_ARGUMENT_EXPRESSION(NotExp, !, bool)
ONE_ARGUMENT_EXPRESSION(NegateExp, -, A)
ONE_ARGUMENT_EXPRESSION(LogExp, log, double)

#define TYPE_CHOISE_4(NAME, ACTION)                            \
if(t1 == BOOL)   {                                             \
	/*printf("%s BOOL\n", #ACTION) ;*/                         \
	return new NAME<bool> (a) ;                                \
}

#define TYPE_CHOISE_5(NAME, ACTION)                            \
if(t1 == INT)   {                                              \
	/*printf("%s INT\n", #ACTION) ;*/                          \
	return new NAME<int32> (a) ;                               \
}                                                              \
if(t1 == DOUBLE)   {                                           \
	/*printf("%s DOUBLE\n", #ACTION) ;*/                       \
	return new NAME<double> (a) ;                              \
}

#define TYPE_CHOISE_6(NAME, ACTION)                            \
if(t1 == INT)   {                                              \
	/*printf("%s INT\n", #ACTION) ;*/                          \
	return new NAME<int32> (a) ;                               \
}                                                              \
if(t1 == DOUBLE)   {                                           \
	/*printf("%s DOUBLE\n", #ACTION) ;*/                       \
	return new NAME<double> (a) ;                              \
}


template<class A>
class IfExpr: public Exp {
  private:
  	A *data ;
  	bool *source1 ;
  	A *source2, *source3 ;
  	Exp *a, *b, *c ;
  public:
  	IfExpr(Exp* a_, Exp* b_, Exp* c_) : a(a_), b(b_), c(c_) {
  		data = new A[TABLESIZE] ;
  		output = Output_elem((void *) data, toType< A >::X) ;
  		source1 = (bool*) a->getOutputElem().column ;
  		source2 = (A*) b->getOutputElem().column ;
  		source3 = (A*) c->getOutputElem().column ;
  	}
  	void doit(int size) {
  		a->doit(size) ;
  		b->doit(size) ;
  		c->doit(size) ;
  		for(int i=0 ; i<size ; i++)
  			data[i] = (source1[i]) ? source2[i] : source3[i] ;
  	}
} ;


Exp* createExp(const Expression &expression, const Output &source) {
	switch(expression.operator_()) {
		case Expression_Operator_COLUMN :
			return new ColumnExp(source, expression.column_id()) ;
		case Expression_Operator_CONSTANT :
			if(expression.has_constant_int32()) {
//				cout << "INT" << endl ;
				return new ConstantExp<int32>(expression.constant_int32()) ;
			}
			else if(expression.has_constant_double()) {
//				cout << "DOUBLE" << endl ;
				return new ConstantExp<double>(expression.constant_double()) ;
			}
			else if(expression.has_constant_bool()) {
//				cout << "BOOL" << endl ;
				return new ConstantExp<bool>(expression.constant_bool()) ;
			}
			else assert(false) ;
		case Expression_Operator_ADD :
		case Expression_Operator_SUBTRACT :
		case Expression_Operator_MULTIPLY :
		case Expression_Operator_LOWER :
		case Expression_Operator_GREATER :
		case Expression_Operator_EQUAL :
		case Expression_Operator_NOT_EQUAL : 
		case Expression_Operator_OR :
		case Expression_Operator_AND :
		case Expression_Operator_FLOATING_DIVIDE : {
			Exp* a = createExp(expression.children(0), source) ;
			Exp* b = createExp(expression.children(1), source) ;
			Type t1 = a->getOutputElem().type ;
			Type t2 = b->getOutputElem().type ;
//			cout << t1 << " " << t2 << endl ;
			switch(expression.operator_()) {
				case Expression_Operator_ADD :      TYPE_CHOISE_1(AddExp, +) ;
				case Expression_Operator_SUBTRACT : TYPE_CHOISE_1(SubtractExp, -) ;
				case Expression_Operator_MULTIPLY : TYPE_CHOISE_1(MultiplyExp, *) ;
				assert(false) ;
				case Expression_Operator_LOWER :     TYPE_CHOISE_2(LowerExp, <) ;
				case Expression_Operator_GREATER :   TYPE_CHOISE_2(GreaterExp, >) ;
				case Expression_Operator_EQUAL :     TYPE_CHOISE_2(EqualExp, ==) ;
				case Expression_Operator_NOT_EQUAL : TYPE_CHOISE_2(NotEqualExp, !=) ;
				assert(false) ;
				case Expression_Operator_OR :   TYPE_CHOISE_3(OrExp, ||) ;
				case Expression_Operator_AND :  TYPE_CHOISE_3(AndExp, &&) ;
				assert(false) ;
				case Expression_Operator_FLOATING_DIVIDE: TYPE_CHOISE_1(DivideExp, /) ;
				assert(false) ;
				default: assert(false) ;
			}
			assert(false) ;
		}
		case Expression_Operator_NOT :
		case Expression_Operator_NEGATE :
		case Expression_Operator_LOG : {
			Exp* a = createExp(expression.children(0), source) ;
			Type t1 = a->getOutputElem().type ;
			switch(expression.operator_()) {
				case Expression_Operator_NOT : TYPE_CHOISE_4(NotExp, !) ;
				assert(false) ;
				case Expression_Operator_NEGATE : TYPE_CHOISE_5(NegateExp, -) ;
				assert(false) ;
				case Expression_Operator_LOG : TYPE_CHOISE_6(LogExp, log) ;
				assert(false) ;
				default: assert(false) ;
			}
		}
		case Expression_Operator_IF : {
			Exp* a = createExp(expression.children(0), source) ;
			Exp* b = createExp(expression.children(1), source) ;
			Exp* c = createExp(expression.children(2), source) ;
			Type t1 = a->getOutputElem().type ;
			Type t2 = b->getOutputElem().type ;
			Type t3 = c->getOutputElem().type ;
			assert(t1 == BOOL && t2 == t3) ;
//			cout << "IF " << endl ;
			switch(t2) {
				case INT :
//					cout << "INT INT" << endl ;
					return new IfExpr<int32> (a,b,c) ;
				case DOUBLE:
//					cout << "DOUBLE DOUBLE" << endl ;
					return new IfExpr<double> (a,b,c) ;
				case BOOL:
//					cout << "BOOL BOOL" << endl ;
					return new IfExpr<bool> (a,b,c) ;
				default: assert(false) ;
			}
		}
		default:
//			printf("Nie zdefiniowane wyrazenie\n") ;
			assert(false) ;
	}
}

class OpCompute: public Op {
  private:
  	Op* source ;
  	vector< Exp* > expression ;
  public:
  	OpCompute(const ComputeOperation &operation) {
//  		printf("tworze OpCompute\n") ;
  		source = createOp(operation.source()) ;
  		int n = operation.expressions().size() ;
//  		cout << "tworze wyrazenia" << endl ;
  		for(int i = 0 ; i<n ; i++) {
  			expression.push_back( createExp(operation.expressions(i), source->getOutput()) ) ;
  			output.push_back(expression.back()->getOutputElem()) ;
  		}
  	}
  	int doit() {
  		int ret = source->doit() ;
  		if(ret == -1) return -1 ;
  		else {
  			for(size_t i=0 ; i<expression.size() ; i++)
  				expression[i]->doit(ret) ;
  			return ret ;
  		}
  	}
} ;

template<class T>
inline int rewrite(void *destination, void* source, bool* conditions, int size) {
	T* dest = (T*) destination ;
	T* src = (T*) source ;
	int next=0 ;
	for(int i=0 ; i<size ; i++)
		if(conditions[i])
			dest[next++] = src[i] ;
	return next ;
}

class OpFilter: public Op {
  private:
  	Op* source ;
  	Exp* expression ;
  	bool* conditions ;
  	Output input ;
  public:
  	OpFilter(const FilterOperation &operation) {
//  		printf("tworze OpFilter\n") ;
  		source = createOp(operation.source()) ;
  		input = source->getOutput() ;
  		
//  		cout << "tworze wyrazenie warunkowe" << endl ;
  		expression = createExp(operation.expression(), input) ;
  		
  		Output_elem elem = expression->getOutputElem() ;
  		assert(elem.type == BOOL) ;
  		conditions = (bool*) elem.column ;
  		
  		// alokacja nowych wskaznikow
  		output = input ;
  		for(size_t i=0 ; i<output.size() ; i++) {
  			switch(output[i].type) {
				case INT:
					output[i].column = malloc(sizeof(int32)*TABLESIZE) ;
					break ;
				case DOUBLE:
					output[i].column = malloc(sizeof(double)*TABLESIZE) ;
					break ;
				case BOOL:
					output[i].column = malloc(sizeof(bool)*TABLESIZE) ;
					break ;
			}
		}
//  		printf("ok\n") ;
  	}
  	int doit() {
  		int size = source->doit() ;
  		if(size == -1) return -1 ;
  		else {
  			expression->doit(size) ;
  			int ret = 0 ;
  			for(size_t i = 0 ; i<output.size() ; i++) {
  				switch(input[i].type) {
  					case INT:
						ret = rewrite<int32>(output[i].column, input[i].column, conditions, size) ;
						break ;
					case DOUBLE:
						ret = rewrite<double>(output[i].column, input[i].column, conditions, size) ;
						break ;
					case BOOL:
						ret = rewrite<bool>(output[i].column, input[i].column, conditions, size) ;
						break ;
				}
			}
			return ret ;
  		}
  	}
} ;

const int MOD = 1000000009 ;
const int BASE = 100000007 ;

struct RowHasher {
  private:
  	int n ;
  public:
  	RowHasher(int n_=0) : n(n_) {} ;
    size_t operator()(void* v) const {
    	long long ret = 0 ;
    	int32* q = (int32*) v ;
    	for(int i=0 ; i<n ; i++) {
    		ret = (BASE*ret + q[i])%MOD ;
    	}
    	return (size_t) ret ;
    }
} ;

struct RowEqual {
  private:
  	int n ;
  public:
  	RowEqual(int n_=0) : n(n_) {} ;
    bool operator()(void* a, void* b) const {
    	int32* q = (int32*) a ;
    	int32* w = (int32*) b ;
    	for(int i=0 ; i<n ; i++)
    		if(q[i] != w[i]) return false ;
    	return true ;
    }
} ;

class OpGroupBy: public Op {
  private:
  	Op* source ;
  	Output by_column ;
  	Output aggregated_column ;
  	bool is_computed_hash_set ;
  	size_t by_size, row_size ;
  	vector< void* > rows ;
  	void *next_row, *end_row ;
  public:
  	OpGroupBy(const GroupByOperation &operation) {
//  		printf("tworze GroupBy\n") ;
  		
  		is_computed_hash_set = false ;
  		by_size = row_size = 0 ;
  		next_row = end_row = 0 ;
  		
  		source = createOp(operation.source()) ;
  		Output in = source->getOutput() ;
  		
  		// przetwarzanie identyfikatorow kolumn po ktorych grupuje
  		int n = operation.group_by_column().size() ;
  		for(int i=0 ; i<n ; i++) {
//  			cout << "numer kolumny group = " << operation.group_by_column(i) << endl ;
  			by_column.push_back( in[operation.group_by_column(i)] ) ;
  			switch(by_column.back().type) {
  				case INT:
  					output.push_back( Output_elem(malloc(sizeof(int32)*TABLESIZE),INT) ) ;
  					break ;
  				case BOOL:
  					output.push_back( Output_elem(malloc(sizeof(bool)*TABLESIZE),BOOL) ) ;
  					break ;
  				default: assert(false) ;
  			}
  			row_size += sizeof(int32) ;
  		}
  		
  		by_size = row_size ;
//  		cout << "by_size = " << by_size << endl ;
  		
  		// przetwarzanie identyfikatorow kolumn po ktorych jest agregacja
  		n = operation.aggregations().size() ;
  		for(int i=0 ; i<n ; i++) {
  			Aggregation ag = operation.aggregations(i) ;
  			switch(ag.type()) {
  				case Aggregation_Type_SUM :
//  					cout << "numer kolumny aggregated = " << ag.aggregated_column() << endl ;
  					aggregated_column.push_back( in[ag.aggregated_column() ]) ;
  					switch(aggregated_column.back().type) {
		  				case INT:
  							output.push_back( Output_elem(malloc(sizeof(int32)*TABLESIZE),INT) ) ;
  							row_size += sizeof(int32) ;
  							break ;
  						case DOUBLE:
  							output.push_back( Output_elem(malloc(sizeof(double)*TABLESIZE),DOUBLE) ) ;
  							row_size += sizeof(double) ;
  							break ;
  						default: assert(false) ;
  					}
  					break ;
  				case Aggregation_Type_COUNT :
  					// 0 to wartownik ktory mowi ze zamiast SUM jest COUNT (!)
//  					cout << "COUNT" << endl ;
  					aggregated_column.push_back( Output_elem(0, INT) ) ;
					output.push_back( Output_elem(malloc(sizeof(int32)*TABLESIZE),INT) ) ;
					row_size += sizeof(int32) ;
  					break ;
  			}
  		}
  		
//  		cout << "row_size = " << row_size << endl ;
  		
  	}
  private:
    void allocate_next_rows() {
    	next_row = malloc(row_size * TABLESIZE) ;
    	rows.push_back(next_row) ;
    	end_row = next_row + row_size * TABLESIZE ;
    }
  public:
  	int doit() {
  		if(!is_computed_hash_set) {
//  			cout << "licze hashmape" << endl ;
  			is_computed_hash_set = true ;
  			
  			typedef unordered_set< void*, RowHasher, RowEqual > RowHashSet ;
  			RowHashSet S(TABLESIZE, RowHasher(by_column.size()) ,
  									RowEqual (by_column.size()) ) ;
  			
//  			cout << "liczba kubelkow: " << S.bucket_count() << endl ;
  			
  			int size ;
  			while( (size = source->doit()) != -1) {
  				for(int i=0 ; i<size ; i++) {
  					if(next_row == end_row) allocate_next_rows() ;
  					int32* r = (int32*) next_row ;
  					for(size_t j=0 ; j<by_column.size() ; j++) {
  						switch(by_column[j].type) {
  							case INT:
  								*r = *(((int32*) by_column[j].column) +i) ;
  								break ;
  							case BOOL:
  								*r = (int32) *(((bool*) by_column[j].column) +i) ;
  								break ;
  							default: assert(false) ;
  						}
  						++r ;
  					}
  					
  					void *q ;
  					pair<RowHashSet::iterator, bool> p = S.insert(next_row) ;
  					if(p.second) { // czyli jest wlozone
  						memset(r, 0, row_size-by_size) ;
  						q = (void*) r ;
  					}
  					else {
  						q = (*p.first) + by_size ;
  					}
  					
  					// wlasciwa agragacja
  					for(size_t j=0 ; j<aggregated_column.size() ; j++) {
  						if(aggregated_column[j].column == 0) { // specjalny przypadek oznaczajacy COUNT
//							cout << "mam COUNT "  ;
  							int32 *wsk = (int32*) q ;
  							(*wsk)++ ;
//							cout << *wsk << endl ;
  							q += sizeof(int32) ;
  						}
  						else {
  							switch(aggregated_column[j].type) {
  								case INT: {
  									int32 *wsk = (int32*) q ;
  									*wsk += *(((int32*) aggregated_column[j].column) +i) ;
  									q += sizeof(int32) ;
  									break ;
  								}
  								case DOUBLE: {
  									double *wsk = (double*) q ;
  									*wsk += *(((double*) aggregated_column[j].column) +i) ;
  									q += sizeof(double) ;
  									break ;
  								}
  								default: assert(false) ;
  							}
  						}
  					}
  					
  					if(p.second) {
	  					next_row += row_size ;
//  						assert(q == next_row) ;
  					}
  				}
  			}
  		}
  		
  		if(rows.empty()) return -1 ;
  		else {
  			void* q = rows.back() ;
  			int size=0 ;
  			while(q != next_row) {
  				void* r = q ;
  				for(size_t j=0 ; j<output.size() ; j++) {
  					switch(output[j].type) {
  						case INT:
  							*(((int32*) output[j].column) + size) = *( (int32*)r ) ;
  							r += sizeof(int32) ;
  							break ;
  						case DOUBLE:
  							*(((double*) output[j].column) + size) = *( (double*)r ) ;
  							r += sizeof(double) ;
  							break ;
  						case BOOL:
  							*(((bool*) output[j].column) + size) = (bool) *( (int32*)r ) ;
  							r += sizeof(int32) ;
  							break ;
  					}
  				}
  				size++ ;
//  				assert(r == q+row_size) ;
  				q += row_size ;
  			}
  			free(rows.back()) ;
  			rows.pop_back() ;
  			if(!rows.empty()) next_row = rows.back() + row_size*TABLESIZE ;
  			return size ;
  		}
  	}
} ;


Op* createOp(const Operation &operation) {
	if(operation.has_scan()) return new OpScan(operation.scan()) ;
	else if(operation.has_compute()) return new OpCompute(operation.compute()) ;
	else if(operation.has_filter()) return new OpFilter(operation.filter()) ;
	else if(operation.has_group_by()) return new OpGroupBy(operation.group_by()) ;
	else {
//		printf("puste! \n") ;
		assert(false) ;
	}
}

int main(int argc, char* argv[]) {
	if(argc < 3) {
		printf("Uposob uzycia: %s numer_testu nazwa_pliku_proto\n", argv[0]) ;
		return 1 ;
	}
	
	int file = open(argv[2], O_RDONLY) ;
	if(file == -1) {
		printf("Blad otwierania pliku\n") ;
		return 1 ;
	}
	
	server = CreateServer(atoi(argv[1])) ;
	
	Operation operation ;
	google::protobuf::io::FileInputStream input(file);
	google::protobuf::TextFormat::Parse(&input, &operation);
	
	Op *op = createOp(operation) ;
	
	Output output = op->getOutput() ;
	
	int ret ;
	while( (ret = op->doit()) != -1 ) {
		for(size_t i=0 ; i<output.size() ; i++) {
			switch(output[i].type) {
				case INT:
					server->ConsumeInts(i, ret, (int32*) output[i].column) ;
					break ;
				case DOUBLE:
					server->ConsumeDoubles(i, ret, (double*) output[i].column) ;
					break ;
				case BOOL:
					server->ConsumeByteBools(i, ret, (bool*) output[i].column) ;
					break ;
			}
		}
	}
	delete server ;
//	printf("\nexit ok\n") ;
	return 0 ;
}
